package com.atliuzu.app.dwd.log;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atliuzu.utils.DateFormatUtil;
import com.atliuzu.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


/**
 * @author w
 * @create 2022-08-19-10:41
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {

        //1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.状态后端
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies
//                .failureRateRestart(10,
//                        Time.of(3L, TimeUnit.DAYS),
//                        Time.of(1L, TimeUnit.MINUTES)));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");


        //3.读取log主题的数据，筛选
        //需要定义kafka消费工具类
        String topic = "topic_log";
        String groupId = "base_log_app";
        DataStreamSource<String> baseDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        SingleOutputStreamOperator<JSONObject> jsonDS = baseDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    System.out.println("脏数据" + value);
                }
            }
        });

        //4.新老用户状态修复
        //按照mid分组
        KeyedStream<JSONObject, String> keyedDS = jsonDS.keyBy(json -> json.getJSONObject("common").getString("mid"));

        //思路将 上次登录时间存入状态
        // 若 isnew ==1   且状态为null   更新状态为当前时间
        //                当前状态不为空且为今天    写入状态
        //                  当前状态不为空且时间不为今天  修改 isnew==0   写入状态
        // 若 isnew ==0   且状态为null   将isnew==1 更新状态为当前时间
        //                  且状态不为空   不修改
        SingleOutputStreamOperator<JSONObject> processDS = keyedDS.process(new ProcessFunction<JSONObject, JSONObject>() {
            ValueState<String> lastDtState;

            @Override
            public void open(Configuration parameters) throws Exception {
                lastDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("base_log", String.class));
            }

            @Override
            public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
                //获取时间和isnew字段
                String lastDt = lastDtState.value();
                String isNew = value.getJSONObject("common").getString("is_new");
                Long ts = value.getLong("ts");
                String curDt = DateFormatUtil.toDate(ts);

                if ("1".equals(isNew)) {
                    if (lastDt == null) {
                        lastDtState.update(curDt);
                    } else {
                        if (!lastDt.equals(curDt)) {
                            isNew = "0";
                            value.getJSONObject("common").put("is_new", isNew);
                        }
                    }
                } else if (lastDt == null) {
                    //更新状态为昨日，这样后面进来的用户都为老用户了
                    String yesterday = DateFormatUtil.toDate(ts - 24 * 60 * 60 * 1000L);
                    lastDtState.update(yesterday);
                }
                out.collect(value);
            }
        });


        //5.使用侧输出流进行分流处理
        OutputTag<String> errorTag = new OutputTag<String>("error") {
        };
        OutputTag<String> startTag = new OutputTag<String>("start") {
        };
        OutputTag<String> displaysTag = new OutputTag<String>("displays") {
        };
        OutputTag<String> actionsTag = new OutputTag<String>("actions") {
        };
        OutputTag<String> videoTag = new OutputTag<String>("appVideo") {
        };


        SingleOutputStreamOperator<String> pageDS = processDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                //尝试获取error字段
                String err = value.getString("err");
                if (err != null) {
                    ctx.output(errorTag, value.toString());
                    value.remove("err");
                }

                //尝试获取appVideo字段
                String video = value.getString("appVideo");
                //尝试获取start字段
                String start = value.getString("start");
                if (start != null) {
                    ctx.output(startTag, value.toString());
                }
                else if(video != null) {
                    ctx.output(videoTag, value.toString());
                }
                else {
                    //必为页面字段
                    JSONObject common = value.getJSONObject("common");
                    Long ts = value.getLong("ts");
                    JSONObject page = value.getJSONObject("page");


                    //尝试获取displays
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null) {
                        //遍历写出
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            display.put("common",common);
                            display.put("ts",ts);
                            display.put("page", page);
                            ctx.output(displaysTag, display.toString());
                        }
                    }

                    //尝试获取actions
                    JSONArray actions = value.getJSONArray("actions");
                    if (actions != null) {
                        //遍历写出
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            action.put("common",common);
                            action.put("ts",ts);
                            action.put("page", page);
                            ctx.output(actionsTag, action.toString());
                        }
                    }


                    //移除曝光和动作数据
                    value.remove("displays");
                    value.remove("actions");

                    out.collect(value.toJSONString());
                }
            }
        });

        //6.提取各个流的数据&写出到Kafka
        DataStream<String> startTagDS = pageDS.getSideOutput(startTag);
        DataStream<String> displaysTagDS = pageDS.getSideOutput(displaysTag);
        DataStream<String> actionsTagDS = pageDS.getSideOutput(actionsTag);
        DataStream<String> errorTagDS = pageDS.getSideOutput(errorTag);
        DataStream<String> videoTagDS = pageDS.getSideOutput(videoTag);


        pageDS.print("Page>>>>");
//        startTagDS.print("Start>>>>");
//        displaysTagDS.print("Display>>>>");
//        actionsTagDS.print("Action>>>>");
//        errorTagDS.print("Error>>>");
       //videoTagDS.print("video>>>");


        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";
        String video_topic = "dwd_traffic_video_log";

        pageDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startTagDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displaysTagDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionsTagDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorTagDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));
        videoTagDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(video_topic));


        //7.执行任务
        env.execute();

    }
}
